package com.bkjk.dbConn;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Set;
import java.util.stream.Stream;

import org.bson.Document;

import com.bkjk.constant.MongoConstant;
import com.bkjk.utils.KettleUtils;
import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoIterable;

/**
 * 使用单例模式
 * 
 * @author liangjie
 *
 */
public class MongoDao {

	private MongoDao() {
		// 创建对象的时候，初始化相关信息
		//init_NonOrIncreCollection(MongoConstant.DB_NAME);\
		initConn();
	}
	
	// 不会自动增加表所创建的长度
	private final static int incre_length = 15;
	private MongoClient mc;
	// 其中events是一个app
	private List<String> increLis = null;
	private List<String> increLisFixed = null;
	//此处采用优先队列，主要是希望能够对采集的表排列，方便与对比
	Queue<String> increaseCollections = new PriorityQueue<String>();
	Queue<String> non_increaseCollections = new PriorityQueue<String>();
	private final static char separator = ',';

	private void initTablesForBoth() {
		increLis = KettleUtils.getIncreTables();
		increLisFixed = KettleUtils.getIncreTablesFixed();

	}
	


	private void initConn() {
		
		mc = McHolder.lazyHolder;
	}

	private static class McHolder{
		
		
		public static MongoClient lazyHolder = getMongoClient();
		
		public static MongoClient getMongoClient(){
			MongoCredential credential = MongoCredential.createScramSha1Credential(MongoConstant.DB_USERNAME, MongoConstant.AUTH_DB_NAME, MongoConstant.DB_PASSWORD.toCharArray());
			ServerAddress serverAddress = new ServerAddress(MongoConstant.MONGO_HOST, MongoConstant.MONGO_PORT);
			List<ServerAddress> addrs = new ArrayList<ServerAddress>();
            addrs.add(serverAddress);
            List<MongoCredential> credentials = new ArrayList<MongoCredential>();
            credentials.add(credential);
            MongoClient mongoClient = new MongoClient(addrs, credentials);
            return mongoClient;
		}
	}
	/**
	 * 列出集合中所有的数据库名
	 * 
	 * @param mc
	 * 
	 * 
	 */
	public void listDB(MongoClient mc) {
		MongoIterable<?> lis = mc.listDatabaseNames();
		MongoCursor<?> iter = lis.iterator();
		while (iter.hasNext()) {
			System.out.println(iter.next().toString());
		}
	}

	/**
	 * 列出mongo中所有集合名称
	 * 
	 * @param mc
	 * @param databaseName
	 */
	public void listCollections(MongoClient mc, String databaseName) {

		initTablesForBoth();

		MongoDatabase database = mc.getDatabase(databaseName);
		MongoIterable<String> cs = database.listCollectionNames();
		MongoCursor<String> iter2 = cs.iterator();

		while (iter2.hasNext()) {
			System.out.println(iter2.next().toString());
		}
	}

	/**
	 * 列出mongo中所有集合名称
	 * 
	 * @param mc
	 * @param databaseName
	 */
	public void init_NonOrIncreCollection(String databaseName) {
		//initConn();
		initTablesForBoth();

		MongoDatabase database = mc.getDatabase(databaseName);
		MongoIterable<String> cs = database.listCollectionNames();
		MongoCursor<String> iter2 = cs.iterator();

		while (iter2.hasNext()) {
			// collectionName = iter2.next().toString();
			// 只要collectionName中包含lis中任何一项的字符
			final String collectionName = iter2.next().toString();
			// 对于list中的每一个，如果collectionName包含该字符串,该filter会留下包含lis中项的值,由于会有一些ID之类的，会自动增加的表名长度肯定大于15
			// 不然events和systemlogs表会包含在内。
			Stream<String> stIncre = increLis.stream()
					.filter(s -> (collectionName.length() > incre_length && collectionName.indexOf(s) > -1));
			
			//此处分为俩个主要是方便左测试，当使用incre时，对nonIncre不产生影响
			Stream<String> stIncreFixed = increLisFixed.stream()
					.filter(s -> (collectionName.length() > incre_length && collectionName.indexOf(s) > -1));
			//一定时非自增表
			if(stIncreFixed.count()<=0){
				non_increaseCollections.add(collectionName);
				
			}
			//当一定大于0时,还需要判断
			else {
				if(stIncre.count()>0){
					increaseCollections.add(collectionName);
				}
			}
			
		}
		// 将结果中的系统索引表去除
		//non_increaseCollections.remove("system.indexes");

		
		//close();
	}

	/**
	 * 获取一个列中的所有列，通过遍历的方式,
	 * 由于无法确定每个列的类型，所以先使用pdm设计出表，然后放入到mysql。
	 * @param collectionName
	 * @param dbName
	 * @return
	 */
	public Set<String> getColumnsInCollection(String collectionName,String dbName){
		MongoDatabase db = mc.getDatabase(dbName);
		MongoCollection<?> mongoCollection = db.getCollection(collectionName);
		FindIterable<?> results = mongoCollection.find();
		MongoCursor<?> r_iter = results.iterator();
		Set<String> set = new HashSet<String>();
		while(r_iter.hasNext()){
			Document doc = (Document) r_iter.next();
			Set<Entry<String,Object>> entrys = doc.entrySet();
			for(Entry<String, Object> entry:entrys){
				//将所有空格替换为_
				
				set.add(entry.getKey());
				//set.add(entry.getKey().replaceAll(" ", "_"));
			}
		}
		return set;
	}
	
	
	public List<String> getColumnfromCollection(String collectionName,String columnName){
		MongoDatabase db = mc.getDatabase(MongoConstant.DB_NAME);
		MongoCollection<Document> collection = db.getCollection(collectionName);
		FindIterable<Document> docs = collection.find();
		List<String> lis = new ArrayList<>();
		for(Document doc:docs){
			lis.add(doc.get(columnName).toString());
		}
		return lis;
	}
	
	public void close() {
		mc.close();
	}

	public String listToString(List<?> list, char separator) {
		StringBuilder sb = new StringBuilder();
		list.stream().forEach(s -> sb.append(s).append(separator));
		return list.isEmpty() ? "" : sb.toString().substring(0, sb.toString().length() - 1);
	}

	public String listToString(Queue<String> queue, char separator) {
		StringBuilder sb = new StringBuilder();
		queue.stream().forEach(s -> sb.append(s).append(separator));
		return queue.isEmpty() ? "" : sb.toString().substring(0, sb.toString().length() - 1);
	}


	public void test() {

		// listDB(mc);
		initConn();
		String databaseName = "countly";

		// listCollections(mc,"countly");
		init_NonOrIncreCollection(databaseName);
		increaseCollections.spliterator();
		increaseCollections.stream().forEach(System.out::println);

		System.out.println("非------------------------------------------------------------------------");
		non_increaseCollections.forEach(System.out::println);
		// listDocument(mc, databaseName, collectionName);
		String increCol = listToString(increaseCollections, separator);
		String non_increCol = listToString(non_increaseCollections, separator);
		System.out.println("放到一起的字符串increCol:" + increCol);
		System.out.println("放到一起的字符串non_increCol:" + non_increCol);
		close();
	}

	
	
	public static final MongoDao getInstance() {    
	       return LazyHolder.INSTANCE;    
	}   
	
	private static class LazyHolder{
		private  static final MongoDao INSTANCE = new MongoDao();
	}

	public Queue<String> getIncreaseCollections() {
		return increaseCollections;
	}

	public void setIncreaseCollections(Queue<String> increaseCollections) {
		this.increaseCollections = increaseCollections;
	}

	public Queue<String> getNon_increaseCollections() {
		return non_increaseCollections;
	}

	public void setNon_increaseCollections(Queue<String> non_increaseCollections) {
		this.non_increaseCollections = non_increaseCollections;
	}
	
	
	public MongoClient getMc() {
		return mc;
	}

	public void setMc(MongoClient mc) {
		this.mc = mc;
	}

	public static void main(String[] args) {
		MongoDao app = MongoDao.getInstance();
		app.initConn();
		app.getColumnsInCollection("cities", "countly");
		
		//app.test();
		
		app.close();
	}

	
}
